//package com.k2data.platform.service.adapters.fixedformat.bolt;
//
//import backtype.storm.task.OutputCollector;
//import backtype.storm.task.TopologyContext;
//import backtype.storm.topology.OutputFieldsDeclarer;
//import backtype.storm.topology.base.BaseRichBolt;
//import backtype.storm.tuple.Tuple;
//
//import com.k2data.platform.ddm.share.common.Configuration;
//import com.k2data.platform.ddm.share.common.Constants;
//import com.k2data.platform.ddm.share.common.ParamNames;
//import com.k2data.platform.ddm.share.schema.service.SchemaService;
//
//import com.k2data.platform.ddm.share.schema.utils.AvroUtil;
//import org.apache.avro.Schema;
//import org.apache.avro.generic.GenericRecord;
//import org.apache.avro.io.*;
//import org.apache.avro.specific.SpecificDatumReader;
//import org.apache.commons.configuration.ConfigurationException;
//
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import redis.clients.jedis.Jedis;
//import storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
//import storm.kafka.bolt.mapper.TupleToKafkaMapper;
//
//import java.io.*;
//import java.util.HashMap;
//import java.util.LinkedHashMap;
//import java.util.Map;
//
//
///**
// * Created by luoqifei on 17-5-3.
// */
//public class RedisAvroBolt extends BaseRichBolt {
//    private static Logger logger = LoggerFactory.getLogger(RedisAvroBolt.class);
//    private Jedis jedis;
//    private OutputCollector _collector;
//    private byte[] configBytes;
//    private Configuration _configuration;
//    private SchemaService schemaService;
//    private String schemaId = "";
//    private Map<String, String> idGroupFields;
//    private Map<String, String> nonIdFieldToTypeMap;
//    private Schema schema ;
//    private GenericRecord record;
//    private final static String k2AdapterMode= "K2_ADAPTER_MODE";
//    private final static String nullStr="";
//    private final static String timestampName = "timestamp";
//    private long timeStamp = 0l;
//    public RedisAvroBolt(byte[] configuration) {
//        configBytes = configuration;
//
//    }
//
//    @Override
//    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
//
//        _collector = collector;
//        idGroupFields = new HashMap<>();
//        nonIdFieldToTypeMap = new HashMap<>();
//        // load configuration
//        _configuration = new Configuration();
//        try {
//            _configuration.fromBytes(configBytes);
//        } catch (ConfigurationException e) {
//            logger.error("Fail to get Configuration from byte[]!");
//            e.printStackTrace();
//            System.exit(1);
//        } catch (IOException e) {
//            logger.error("Fail to get Configuration from byte[]!");
//            e.printStackTrace();
//            System.exit(1);
//        }
//        if (logger.isInfoEnabled()) {
//            logger.info("user configuration: {}", _configuration.toString());
//        }
//
//        // load configuration from zookeeper if not set K2_ADAPTER_MODE to "local"
//        if (!System.getenv().containsKey(k2AdapterMode) || !System.getenv().get(k2AdapterMode).equals("local")) {
//            String zkUrl = _configuration.getString(ParamNames.ZOOKEEPER_URL);
//            if (zkUrl == null || zkUrl.equals("")) {
//                logger.error("Not set ZOOKEEPER_URL env!");
//                System.exit(1);
//            }
//        }
//        // init jedis client
//        String redisUrlList = _configuration.getString(ParamNames.REDIS_URL_LIST,Constants.REDIS_URL_LIST);
//        String[] redisUrl = redisUrlList.split(",");
//        jedis = new Jedis(redisUrl[0].substring(0,redisUrl[0].indexOf(":")-1),Integer.valueOf(redisUrl[0].substring(redisUrl[0].indexOf(":")+1)));
//        // start schemaService
//        try {
//            if (schemaService == null) {
//                schemaService = new SchemaService(_configuration);
//            }
//            schemaService.start();
//            // sync with zk when preparing bolt task
//            schemaService.reload();
//        } catch (Exception e) {
//            logger.error("Fail to start schemaService: {}", e.getMessage());
//            e.printStackTrace();
//            System.exit(1);
//        }
//    }
//
//    public GenericRecord  deserializeAvroRecord(byte[] recordBytes,Schema schema) throws IOException {
//        ByteArrayInputStream inputStream = new ByteArrayInputStream(recordBytes);
//        Decoder decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
//        DatumReader<CharSequence> datumReader = new SpecificDatumReader<CharSequence>(schema);
//        GenericRecord record2 = (GenericRecord) datumReader.read(null, decoder);
//        return record2;
//    }
//    public void saveToRedis(String noIdFields,String asset,long timestamp){
//        //asset=fg+idFiled
//        noIdFields = noIdFields.substring(0,noIdFields.length()-1);
//        asset = asset.substring(0,asset.length()-1);
//        System.out.println("%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% ");
//        System.out.println("idx fg&idField = "+Constants.Z_KEY_PREFIX+asset);
//        System.out.println("val fg&idField = "+Constants.H_KEY_PREFIX+asset);
//        System.out.println("timestamp = "+timestamp);
//        System.out.println("noIdFields = "+noIdFields);
//
//        String id = jedis.get(asset);
//        if(id == null){
//            id="1";
//        }
//        if(Long.valueOf(id)>100000000L){
//            id="1";
//        }
//        String globId = String.valueOf(Long.valueOf(id));
//        System.out.println("id : "+globId);
//        jedis.zadd(Constants.Z_KEY_PREFIX+asset,timestamp,globId);
//        jedis.hset(Constants.H_KEY_PREFIX+asset,globId,noIdFields);
//        jedis.set(asset,globId);
//    }
//    public LinkedHashMap<String,String> getSensorId2SensorSysId(Schema schema) {
//        LinkedHashMap<String, String> sensorId2SensorSysId = new LinkedHashMap<>();
//        if(schema == null)
//            return null;
//        for(Schema.Field field : schema.getFields()) {
//            if(AvroUtil.isSensorField(field)) {
//                String sensorId = null;
//                for(String aliase : field.aliases()) {
//                    sensorId = aliase;
//                    break;
//                }
//                if(sensorId != null) {
//                    sensorId2SensorSysId.put(sensorId, field.name());
//                }
//            }
//        }
//        return sensorId2SensorSysId;
//    }
//    public LinkedHashMap<String, String> getIdFieldId2IdFieldSysId(Schema schema) {
//        LinkedHashMap<String, String> idFieldId2IdFieldSysId = new LinkedHashMap<>();
//        if(schema == null)
//            return null;
//        for(Schema.Field field : schema.getFields()) {
//            if(AvroUtil.isIdField(field)) {
//                String idFieldId = null;
//                for(String aliase : field.aliases()) {
//                    idFieldId = aliase;
//                    break;
//                }
//                if(idFieldId != null) {
//                    idFieldId2IdFieldSysId.put(idFieldId, field.name());
//                }
//            }
//        }
//        return idFieldId2IdFieldSysId;
//    }
//    @Override
//    public void execute(Tuple input) {
//        byte[] tupleKey = input.getBinaryByField(Constants.KAFKA_DEFAULT_FIELD_KEY);
//        try {
//            schemaId = new String(tupleKey, Constants.CONF_DEFAULT_ENCODING);
//        } catch (UnsupportedEncodingException e) {
//            e.printStackTrace();
//        }
//        schemaId=schemaId.substring(0,schemaId.length()-5);
//        schema = schemaService.getSchema(schemaId);
//        if(schema == null){
//
//        }
//        idGroupFields.clear();
//        nonIdFieldToTypeMap.clear();
//        idGroupFields = getIdFieldId2IdFieldSysId(schema);
//        nonIdFieldToTypeMap = getSensorId2SensorSysId(schema);
//
//        byte[] tupleMessage = input.getBinaryByField(Constants.KAFKA_DEFAULT_FIELD_MESSAGE);
//        StringBuilder fgIdAndIdFiled = new StringBuilder(schemaId+Constants.FGID_IDFIELD_CONNECTOR);
//        StringBuilder noIdFieldValues = new StringBuilder();
//        try {
//
//            record = deserializeAvroRecord(tupleMessage,schema);
//            timeStamp = (Long)record.get(timestampName);
//            for (String fieldId : idGroupFields.keySet()) {
//                if(record.get(idGroupFields.get(fieldId)) != null){
//                    fgIdAndIdFiled.append(record.get(idGroupFields.get(fieldId)));
//                    fgIdAndIdFiled.append(Constants.IDFIELDS_CONNECTOR);
//                }else {
//                    logger.error("the avro record, fgId="+schemaId+" idField name is "+fieldId+" ,it's value null.");
//                }
//            }
//            for(String nonIdFieldSid : nonIdFieldToTypeMap.keySet()){
//                if(record.get(nonIdFieldToTypeMap.get(nonIdFieldSid)) != null){
//                    noIdFieldValues.append(record.get(nonIdFieldToTypeMap.get(nonIdFieldSid)));
//                }else {
//                    noIdFieldValues.append(nullStr);
//                }
//                noIdFieldValues.append(Constants.FIELD_VALUES_CONNECTOR);
//            }
//            saveToRedis(noIdFieldValues.toString(),fgIdAndIdFiled.toString(),timeStamp);
//            record = null;
//            schema = null;
//            _collector.ack(input);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//    }
//    @Override
//    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
//
//    }
//}
